package main

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"logs/config"
	"logs/db"
	"os"
	"strings"
	"sync"
	"time"
)

type LogLine struct {
	Row  int
	Line string
}

type LogContext struct {
	Ts         string
	Row        int
	Level      string
	TraceId    string
	ThreadName string
	ClassName  string
	Context    string
}

var gw sync.WaitGroup

func batchInsert(bachLog []LogContext, appName string, nodeName string) {
	insertSql := fmt.Sprintf("insert into `go_logs`.`log_%v_%v` using `go_logs`.`log` tags ('%v', '%v')", appName, nodeName, appName, nodeName)
	values := make([]string, 0)
	for _, val := range bachLog {
		parse, _ := time.Parse("2006-01-02 15:04:05.999", val.Ts)
		parse = parse.Add(time.Duration(val.Row%1000) * time.Microsecond)
		value := fmt.Sprintf("('%v','%v','%v','%v','%v','%v')",
			parse.Format("2006-01-02 15:04:05.999999"),
			val.Level,
			val.TraceId,
			val.ThreadName,
			val.ClassName,
			val.Context,
		)
		values = append(values, value)
	}
	insertSql = fmt.Sprintf("%v values %v", insertSql, strings.Join(values, " "))
	fmt.Println(insertSql)
	exec, err := db.TaosDb.Exec(insertSql)
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println(exec.RowsAffected())
}

func transferLog(logLineChan chan LogLine, appName string, nodeName string) {
	batchSize := 200
	batch := make([]LogContext, 0, batchSize)
	for logLine := range logLineChan {
		split := strings.Split(logLine.Line, "|")
		logContext := LogContext{
			split[0],
			logLine.Row,
			split[1],
			split[2],
			split[3],
			split[4],
			strings.ReplaceAll(strings.ReplaceAll(split[5], `"`, `\"`), `'`, `\'`),
		}
		batch = append(batch, logContext)
		if len(batch) == batchSize {
			batchInsert(batch, appName, nodeName)
			batch = make([]LogContext, 0, batchSize)
		}
	}
	if len(batch) > 0 {
		batchInsert(batch, appName, nodeName)
	}
	gw.Done()
}

func sourceFile(logLineChan chan LogLine, logFilePath string) {
	file, err := os.OpenFile(logFilePath, os.O_RDONLY, 0655)
	if err != nil {
		log.Fatalln("日志文件打开失败:", err)
	}
	defer func(file *os.File) {
		err := file.Close()
		if err != nil {
			fmt.Println("Error close file:", err)
		}
	}(file)
	reader := bufio.NewReader(file)
	row := 0
	for {
		line, err := reader.ReadString('\n')
		if line != "" && !strings.Contains(line, "|") {
			continue
		}
		if err == io.EOF {
			break
		}
		row++
		logLineChan <- LogLine{row, strings.Trim(line, "\r\n")}
	}
	close(logLineChan)
	gw.Done()
}

func main() {
	start := time.Now().UnixMilli()

	for _, logMeta := range config.Conf.LogMetas {
		if !logMeta.Enable {
			continue
		}
		var logLineChan = make(chan LogLine, 1024)

		gw.Add(1)
		go sourceFile(logLineChan, logMeta.LogFilePath)

		for range 10 {
			gw.Add(1)
			go transferLog(logLineChan, logMeta.AppName, logMeta.NodeName)
		}
	}

	gw.Wait()
	fmt.Printf("用时=>%dms\n", time.Now().UnixMilli()-start)
}
